import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/10/11 09:11
 */
public class Join_3 extends BaseSQLApp {
    public static void main(String[] args) {
        new Join_3().init(4000, 1, "Join_1");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table a(" +
                            "id int, " +
                            "name string)" + SQLUtil.getKafkaSource("a", "Join_1"));
        
        tEnv.executeSql("create table b(" +
                            "id int, " +
                            "age int)" + SQLUtil.getKafkaSource("b", "Join_1"));
        
        
        // 给状态设置一个 ttl
        //tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        
        Table result = tEnv
            .sqlQuery("select " +
                          "a.id, " +
                          "name, " +
                          "age " +
                          "from a " +
                          "left join b on a.id=b.id");
        
        
        tEnv.executeSql("create table ab(" +
                            " id int, " +
                            " name string, " +
                            " age int " +
                            ")with(" +
                            " 'connector' = 'kafka', " +
                            " 'topic' = 'ab2', " +
                            " 'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            " 'format' = 'csv'" +
                            ")");
        
        result.executeInsert("ab");
        //tEnv.executeSql("insert into ab select * from " + result);
        
    }
}
/*
join:
    默认情况下, 会把数据永远存在状态中
    
left join
    如果右表先来, 则正常的每条数据都是新增
    
    如果左表先来, 会有一条结果是 左表有数据, 右表的字段用 null 补齐, 现在如果右表来了,在先删除,在再新增.以后再来都是新增
    
    对左表来说, ttl 表示的是 idle 的时间之后会清空数据
    
kafka连接器:
    普通的 kafka 只能写入新增的数据. 如果有更新和删除是无法写入的.
    
    左连接有删除(-d), 所以无法写入到普通的 kafka 中

upset kafka
    向upset kafka写数据的时候, 如果这次写的数据的 key 不存在, 就认为是新增
    如果 这次的 key 写入的时候 value 是 null, 就表示删除前面相同 key 的数据
    
    如果这次的 key 存在, value 不是 null, 表示是更新
    
 */